package org.example.Model_Prediction;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.example.ch.InitKafka;
import org.example.ch.MyPropertiesUtil;
import org.example.ch.getProperties;
import org.example.develop.HealthAssessmentModelUDF;
import org.example.develop.StringUtilsUDF;
import org.example.utils.MysqlUtils_3;
import org.example.utils.PrepareDimTable;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Properties;

import static org.example.Model_Prediction.QYBLQ_JsonProcessor.unionHealthAssessment;


/**
 * @author chen
 * @version 1.0
 * @description: TODO
 * @date 2023/3/15 10:05
 */

public class QYBLQ {
    private static final String CONFIGPATHS = "config.properties";
    public static SparkSession ss = null;
    public static JavaSparkContext sc = null;
    public static SQLContext sqlContext = null;

    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("qyblqmx")
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .set("spark.driver.allowMultipleContexts", "true")
                .set("spark.hadoop.validateOutputSpecs", "false")
                .set("hive.mapred.supports.subdirectories", "true")
                .set("spark.streaming.backpressure.enabled", "true")
                .set("spark.streaming.kafka.maxRatePerPartition", "60")
                .set("spark.default.parallelism", "100")
                .set("spark.sql.shuffle.partitions", "20")
                .set("mapreduce.input.fileinputformat.input.dir.recursive", "true")
                .set("spark.scheduler.listenerbus.eventqueue.capacity", "100000")
                .set("spark.streaming.kafka.consumer.cache.enabled", "false")
                .set("spark.debug.maxToStringFields", "100");
        sc = new JavaSparkContext(conf);
        sc.setLogLevel("ERROR");
        JavaStreamingContext jsc = new JavaStreamingContext(sc, new Duration(1000 * 60 * 5));
        sqlContext = new SQLContext(sc);
        ss = SparkSession.builder().appName("odstcmsEvent").getOrCreate();

        //注册UDF函数
        ss.udf().register("HealthAssessmentModelUDF", new HealthAssessmentModelUDF(), DataTypes.StringType);
        //tiancy: `t_phm_sqwx` 表中 MXMC、MXMS 字段通过逗号分隔,分隔后的子串进行去重操作.
        ss.udf().register("distinctString", new UDF1<String, String>() {
            public String call(String input) {
                return StringUtilsUDF.distinctString(input);
            }
        }, DataTypes.StringType);

        Properties mysqlProperties_main2 = new Properties();

        mysqlProperties_main2.setProperty("user", MyPropertiesUtil.getPropertiesKey("mysql_user", "config.properties"));
        mysqlProperties_main2.setProperty("password", MyPropertiesUtil.getPropertiesKey("mysql_pwd", "config.properties"));
        mysqlProperties_main2.setProperty("driver", MyPropertiesUtil.getPropertiesKey("mysql_driver", "config.properties"));
        //==========================================================================================================================
        PrepareDimTable.ps_cx_dllx_lj_psd(ss);          //所有车
        PrepareDimTable.Init_JXDT(ss);                  //检修动态
        PrepareDimTable.Init_ZJXC(ss);                  //最近修程
        // |s_train_id|loco_type|loco_no|cksj               |train_status|
        MysqlUtils_3.loadTrainStatusFromPhoenix(ss, "in('HXD1D','HXD1C')", "perAEI"); // aei数据


        //查找车辆最后一次C5\C6修日期（表-》检修动态）
        String sql_jxdt = "select concat(cxjc,'-',ch) as S_TRAIN_ID,max(JGRQ) as JXRQ from (select * from jxdt where RC in('C4修','C5修','C6修')) group by cxjc,ch ";
        ss.sql(sql_jxdt).createOrReplaceTempView("jxdt_mid");


        //模型字典
        ss.read()
                .jdbc(getProperties.getPropertiesKey("mysql-url", CONFIGPATHS), "t_phm_mxzd", mysqlProperties_main2)
                .createOrReplaceTempView("t_phm_mxzd");
        //履历
        ss.read()
                .jdbc(getProperties.getPropertiesKey("mysql-url", CONFIGPATHS), "dim_jmis_jcda_jcll_1", mysqlProperties_main2)
                .createOrReplaceTempView("dim_jmis_jcda_jcll_1");
        ss.read()
                .jdbc(getProperties.getPropertiesKey("mysql-url", CONFIGPATHS), "dim_jmis_t_jcbm_jcgxzd", mysqlProperties_main2)
                .createOrReplaceTempView("dim_jmis_t_jcbm_jcgxzd");
        ss.read()
                .jdbc(getProperties.getPropertiesKey("mysql-url", CONFIGPATHS), "dim_jmis_t_jxgl_jxjld_jbxx", mysqlProperties_main2)
                .createOrReplaceTempView("jxjld_jbxx");


        /**
         * @description: 出库置为A中AEI逻辑：当当前时间-AEIcksj<=20分钟   cksj为AEI出库时间，否则rksj+2H为AEI出库时间
         * @date 2023/6/9 15:07
         * @version 1.0
         */

        String sqlc = "select concat(aeic.loco_type,'-',aeic.loco_no) as s_train_id," +
                "gxzd.LBJBM as PJFLBM," +
                "gxzd.WZMC as WZ," +
                "CONCAT(\"{\\\"模型简称\\\":\\\"\",mxzd.MXMC_JC,\"\\\",\\\"模型结果\\\":\\\"\",'A',\"\\\",\\\"模型结果描述\\\":\\\"\",'-',\"\\\",\\\"模型转义\\\":\\\"\",'良好',\"\\\"}\") as PGCSZB," +
                "'A' as ZHPJ_ZTZL from " +
                // |s_train_id|loco_type|loco_no|cksj               |train_status|
                "(select * from perAEI where loco_type in ('HXD1D','HXD1C') and train_status = '0' and (unix_timestamp(now()) - unix_timestamp(cksj))<300 ) aeic " +
                "left join (select * from dim_jmis_t_jcbm_jcgxzd where LBJBM = 'PJ00003') gxzd on gxzd.SYCX =  aeic.loco_type " +
                "left join (select * from t_phm_mxzd where BJPJM = 'PJ00003') mxzd on mxzd.BJPJM = gxzd.LBJBM";         //取出刚出站1分钟内的数据  并将评估结果置为A
/*        ss.sql(sqlc).write()
                .format("org.apache.spark.sql.execution.datasources.jdbc2")
                .options(objectObjectHashMap)
                .save();*/
        ss.sql(sqlc).write().mode("OverWrite").format("org.apache.phoenix.spark")
                .option("dirver", "org.apache.phoenix.jdbc.PhoenixDriver")
                .option("zkUrl", "jdbc:phoenix:JWGL-YY-T141149,JWGL-YY-T141150,JWGL-YY-T141151,JWGL-YY-T141152,JWGL-YY-V139150,JWGL-YY-V139151,JWGL-YY-V139152,JWGL-YY-V139153:2181:/hbase-unsecure")
                .option("table", "LOCOMOTIVE_ADS.PARTS_LABEL").save();

        //计算部件走向数据
        //1、计算累计走行（取整车累计走行）
        ss.read()
                .jdbc(getProperties.getPropertiesKey("mysql-url", CONFIGPATHS), "dim_jmis_jcda_t_jxglzxgl_v", mysqlProperties_main2)
                .createOrReplaceTempView("dim_jmis_jcda_t_jxglzxgl_v");
        ss.sql("select concat(a.cx,'-',a.ch) as S_TRAIN_id,round(LJZX) as LJZX from dim_jmis_jcda_t_jxglzxgl_v a")
                .createOrReplaceTempView("zx1");

        //2、计算高级修后走行（C5/C6都算高级修）
        ss.read()
                .jdbc(getProperties.getPropertiesKey("mysql-url", CONFIGPATHS), "dim_jmis_jcda_zxgl_day", mysqlProperties_main2)
                .createOrReplaceTempView("dim_jmis_jcda_zxgl_day");
        String sqlgj = "SELECT A.S_TRAIN_ID,round(sum(A.ZXGL_DAY),0) as LJZX FROM (select jxdt_mid.S_TRAIN_ID as S_TRAIN_ID," +
                "zx.ZXGL_DAY as ZXGL_DAY from " +
                "jxdt_mid left join dim_jmis_jcda_zxgl_day zx " +
                "on jxdt_mid.S_TRAIN_ID = concat(zx.CX,'-',zx.CH) and zx.REPOT_DATE > jxdt_mid.JXRQ) A GROUP BY S_TRAIN_ID";
        ss.sql(sqlgj).createOrReplaceTempView("zx2");

        //3、计算最近修后走行
        String sqlxh = "select A.S_TRAIN_ID,round(sum(A.ZXGL_DAY),0) as LJZX from (select " +
                "concat(zjxc.cxjc,'-',zjxc.ch) as S_TRAIN_ID," +
                "zx.ZXGL_DAY as ZXGL_DAY " +
                "from zjxc left join dim_jmis_jcda_zxgl_day zx " +
                "on concat(zx.CX,'-',zx.CH) = concat(zjxc.cxjc,'-',zjxc.ch) and zx.REPOT_DATE > zjxc.KCRQ) A GROUP BY S_TRAIN_ID";
        ss.sql(sqlxh).createOrReplaceTempView("zx3");


        HashMap<String, Object> kafkaParams = InitKafka.init("TCMSevent_QYBLQ111");
        String[] tcms_event_topics = "DWD_CMD_TCMS_FAULT".split(",");
        Collection<String> topic1 = Arrays.asList(tcms_event_topics);
        //消费数据
        JavaInputDStream<ConsumerRecord<String, String>> TCMSStream = KafkaUtils.createDirectStream(
                jsc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.<String, String>Subscribe(topic1, kafkaParams)

        );
        TCMSStream.map(new Function<ConsumerRecord<String, String>, String>() {
            @Override
            public String call(ConsumerRecord<String, String> stringStringConsumerRecord) throws Exception {
                return stringStringConsumerRecord.value();
            }
        }).foreachRDD(new VoidFunction<JavaRDD<String>>() {
            @Override
            public void call(JavaRDD<String> stringJavaRDD) throws Exception {

                if (!stringJavaRDD.isEmpty()) {
                    ss.read().json(stringJavaRDD).createOrReplaceTempView("tcmsEvent");

                    // 这里需要把aei中状态字段以及train_status_time字段传下去
                    // 这里还需要注册修改后aei的临时表 train_status
                    String sql1 = "select te.*, jcll.pjxlh, jcll.pjdjm, aei1.cksj,aei1.train_status\n" +
                            "from (\n" +
                            "       select *, 'TCMS' as gzly, '-' as yyzl, '-' as yfyy, '模型诊断信息' as zdlb, '-' as gzm\n" +
                            "       from tcmsevent\n" +
                            "     ) te\n" +
                            "     left join dim_jmis_jcda_jcll_1 jcll\n" +
                            "     on te.s_train_id = jcll.s_train_id and te.phm_bjwz = jcll.wz and te.bjpjm = jcll.pjflbm\n" +
                            "     left join (select * from perAEI ) aei1 on te.s_train_id = aei1.s_train_id";


                    Dataset<Row> json = ss.sql(sql1).filter("cksj is not null or cksj !=''").toDF().na().
                            fill("-", ss.sql(sql1).schema().fieldNames());

                    //测完
                    /**
                     * @description: TODO
                     * @date 2023/9/14 9:08
                     * @version 1.0
                     *
                     * HXD1C-0001 pjoooo3   1   MXMC   周信息    PGJG  C
                     * HXD1C-0001 pjoooo3   1   MXMC2   周信息    PGJG C
                     * HXD1C-0001 pjoooo3   1   MXMC3   周信息    PGJG D
                     *
                     * HXD1C-0001 pjoooo3   1    PGJG  D
                     *
                     */

                    unionHealthAssessment(ss, json);//得到健康评估结果

                    ss.sql("select\n" +
                            "\tb.s_train_id,\n" +
                            "\tb.BJWZ,\n" +
                            "\tcase\n" +
                            "\t\twhen ((b.BJWZ = '1'\n" +
                            "\t\tand b.mid_pgjg IN ('123','132','213','231','312','321')\n" +
                            "\t\tand b.pgjg = 'B')\n" +
                            "\t\tor b.pgjg = 'C') then 'C'\n" +
                            "\t\twhen b.pgjg = 'A' then 'A'\n" +
                            "\t\telse b.pgjg\n" +
                            "\tend as pgjg\n" +
                            "from\n" +
                            "\t(\n" +
                            "\tselect\n" +
                            "\t\ta.s_train_id, a.BJWZ, a.PGJG, replace(data,\n" +
                            "\t\t',',\n" +
                            "\t\t'') as mid_pgjg\n" +
                            "\tfrom\n" +
                            "\t\t(\n" +
                            "\t\tselect\n" +
                            "\t\t\tS_TRAIN_ID, PGJG, BJWZ, CONCAT_WS(',', COLLECT_LIST(ZXX) over (partition by S_TRAIN_ID)) as data\n" +
                            "\t\tfrom\n" +
                            "\t\t\ttest_mx\n" +
                            "\t\twhere\n" +
                            "\t\t\tBJWZ = '1'\n" +
                            "\t\t\tand BJPJM = 'PJ00003')a)b\n" +
                            "union all \n" +
                            "select\n" +
                            "\tb.s_train_id,\n" +
                            "\tb.BJWZ,\n" +
                            "\tcase\n" +
                            "\t\twhen ((b.BJWZ = '2'\n" +
                            "\t\tand b.mid_pgjg IN ('456','465','546','564','654','645')\n" +
                            "\t\tand b.pgjg = 'B')\n" +
                            "\t\tor b.pgjg = 'C') then 'C'\n" +
                            "\t\twhen b.pgjg = 'A' then 'A'\n" +
                            "\t\telse b.pgjg\n" +
                            "\tend as pgjg\n" +
                            "from\n" +
                            "\t(\n" +
                            "\tselect\n" +
                            "\t\ta.s_train_id, a.BJWZ, a.PGJG, replace(data,\n" +
                            "\t\t',',\n" +
                            "\t\t'') as mid_pgjg\n" +
                            "\tfrom\n" +
                            "\t\t(\n" +
                            "\t\tselect\n" +
                            "\t\t\tS_TRAIN_ID, PGJG, BJWZ, CONCAT_WS(',', COLLECT_LIST(ZXX) over (partition by S_TRAIN_ID)) as data\n" +
                            "\t\tfrom\n" +
                            "\t\t\ttest_mx\n" +
                            "\t\twhere\n" +
                            "\t\t\tBJWZ = '2'\n" +
                            "\t\t\tand BJPJM = 'PJ00003')a)b").distinct().createOrReplaceTempView("ztpg");
                    //评估参数结果


                    ss.sql("select m.S_TRAIN_ID,m.WZMC," +
                                    "CONCAT(\"{\\\"模型ID\\\":\\\"\", id, \"\\\",\\\"模型简称\\\":\\\"\",MXMC_JC,\"\\\",\\\"模型结果\\\":\\\"\",MX_PGJG ,\"\\\",\\\"模型转义\\\":\\\"\",mxzy,\"\\\"}\") as resu from " +
                                    "(select ps.S_TRAIN_ID,a.WZMC,mxzd.mxmc_jc,MXZD.ID as id,case when (mx.PGJG is null or mx.PGJG = '' ) then 'A' else mx.PGJG  end as MX_PGJG," +
                                    "CASE WHEN mx.jgms IS NULL THEN '' ELSE mx.jgms END AS MX_JGMS," +
                                    "case when (mx.PGJG is null or mx.PGJG='A') then '正常' when mx.PGJG='C' THEN '异常' else '故障' end as mxzy " +
                                    " from (select * from ps_cx_dllx_lj_jwd where cx in ('HXD1D','HXD1C')) ps left join " +
                                    "(select SYCX,LBJBM,WZMC from dim_jmis_t_jcbm_jcgxzd where SYCX in ('HXD1D','HXD1C') AND LBJBM = 'PJ00003') a  on ps.cx = a.SYCX " +
                                    "left join (select * from t_phm_mxzd order by id asc) mxzd on a.SYCX = mxzd.syjx AND a.LBJBM = mxzd.BJPJM " +
                                    "left join test_mx MX ON MXZD.ID = MX.MXMC and ps.S_TRAIN_ID = MX.S_TRAIN_ID and a.WZMC = mx.bjwz) m ")
                            .distinct().dropDuplicates("S_TRAIN_ID", "WZMC", "resu").createOrReplaceTempView("pgcs_mid");

                    ss.sql("select S_TRAIN_ID,WZMC,concat_ws('&',collect_list(resu) over(partition by S_TRAIN_ID,WZMC)) as data from pgcs_mid")
                            .distinct().createOrReplaceTempView("pgcs");


                    //检修信息--非常规（以车取数据）
                    String sql = "select m.S_TRAIN_ID," +
                            "m.BJPJM," +
                            "m.wz," +
                            "concat_ws('&',collect_list(resu) over(partition by m.S_TRAIN_ID,m.wz order by m.bm)) as data from " +
                            "(select b.S_TRAIN_ID,b.BJPJM,b.bm,case when b.SCWZ is null then b.xcwz else b.SCWZ end as wz,CONCAT(\"{\\\"部件状态\\\":\\\"\",bj_status,\"\\\",\\\"生产厂家\\\":\\\"\",SCCJ,\"\\\",\\\"检修段\\\":\\\"\",CXDW,\"\\\",\\\"日期\\\":\\\"\",JXRQ_JS,\"\\\",\\\"本次走行公里\\\":\\\"\",BCZXGL,\"\\\",\\\"下车型号\\\":\\\"\",XCXH,\"\\\",\\\"下车位置\\\":\\\"\",XCWZ,\"\\\",\\\"下车日期\\\":\\\"\",XCRQ,\"\\\",\\\"上车型号\\\":\\\"\",SCXH,\"\\\",\\\"上车位置\\\":\\\"\",SCWZ,\"\\\",\\\"上车日期\\\":\\\"\",SCRQ,\"\\\"}\") as resu\n" +
                            " from \n" +
                            "((select \n" +
                            "0 as bm,\n" +
                            "ps.S_TRAIN_ID,\n" +
                            "'牵引变流器' as BJMC,\n" +
                            "'PJ00003' as BJPJM,\n" +
                            "case when SCCJ is null then '-' else SCCJ end as SCCJ,\n" +
                            "date_format(ZZRQ,'yyyy-MM-dd'),\n" +
                            "'新造' as bj_status,\n" +
                            "case when CHANGETIME is null then '-' else date_format(CHANGETIME,'yyyy-MM-dd') end as JXRQ_JS,\n" +
                            "'-' as CXDW,\n" +
                            "'-' as BCZXGL,\n" +
                            "gx.wzmc as SCWZ,\n" +
                            "ps.S_TRAIN_ID as SCXH,\n" +
                            "case when CHANGETIME is null then '-' else date_format(CHANGETIME,'yyyy-MM-dd') end as SCRQ,\n" +
                            "'-' as XCXH,\n" +
                            "'-' as XCRQ,\n" +
                            "'-' as XCWZ \n" +
                            "from \n" +
                            "(select S_TRAIN_ID,CX,CH from ps_cx_dllx_lj_jwd where cx in ('HXD1D','HXD1C')) ps \n" +
                            "left join \n" +
                            "(select * from dim_jmis_t_jcbm_jcgxzd where LBJBM = 'PJ00003' and sycx in ('HXD1D','HXD1C')) gx on ps.cx = gx.sycx\n" +
                            "left join \n" +
                            "(select * from dim_jmis_jcda_jcll_1 where CXM in ('HXD1D','HXD1C') and PJFLBM = 'PJ00003') ll on ps.cx=ll.cxm and ps.ch = ll.ch and gx.LBJBM = ll.PJFLBM and gx.wzmc = ll.wz)\n" +
                            "union all \n" +
                            "select case when RC = 'C4修' THEN 1 WHEN RC = 'C5修' THEN 2 ELSE 3 END AS bm,\n" +
                            "\tconcat(SCCX, '-', SCCH) as S_TRAIN_ID,\n" +
                            "\tcase when FLMC is null then '-' else FLMC end as BJMC,\n" +
                            "\tcase when LBJBM is null then '-' else LBJBM end as BJPJM,\n" +
                            "\tcase when ZZCJ is null then '-' else ZZCJ end as SCCJ,\n" +
                            "\tcase when CCRQ is null then '-' else date_format(CCRQ,'yyyy-MM-dd') end as ZZRQ,\n" +
                            "\tconcat(RC, RT) as BJ_STATUS,\n" +
                            "\tcase when JXRQ_JS is null then '-' else date_format(JXRQ_JS,'yyyy-MM-dd') end as JXRQ_JS,\n" +
                            "\tcase when CXDW is null then '-' else CXDW end as CXDW,\n" +
                            "\tcase when BCZXGL is null then '-' else BCZXGL end as BCZXGL,\n" +
                            "\tcase when SCWZ is null then '-' else SCWZ end as SCWZ,\n" +
                            "\tconcat(SCCX, '-', SCCH) as SCXH,\n" +
                            "\tcase when UPDATE_TIME is null then '-' else date_format(UPDATE_TIME,'yyyy-MM-dd') end as SCRQ,\n" +
                            "\tconcat(XCCX, '-', XCCH) as XCXH,\n" +
                            "\tcase when UPDATE_TIME is null then '-' else date_format(UPDATE_TIME,'yyyy-MM-dd') end as XCRQ,\n" +
                            "\tcase when XCWZ is null then '-' else XCWZ end as XCWZ from jxjld_jbxx jj\n" +
                            "where\n" +
                            "\tSCCX in ('HXD1D','HXD1C') \n" +
                            "\tand LBJBM = 'PJ00003' and RECORD_STATUS = '0') b ) m ";
                    ss.sql(sql).distinct().createOrReplaceTempView("jxxx");

                    //最终结果
                    String sqlre = "SELECT " +
                            "uuid() as ID," +
                            "ps.s_train_id AS S_TRAIN_ID," +
                            "ps.CX AS CX," +
                            "ps.CX_ID AS CX_ID," +
                            "ll.PJDJM as PJDJM," +
                            "'PJ00003' as PJFLBM," +
                            "'01220402' as BJBM," +
                            "case when ll.PJFLM is null then '牵引变流器' else ll.PJFLM end as PJFLM," +
                            "ll.PJXLH as PJXLH," +
                            "ll.gzxh as GGXH_TH," +
                            "date_format(ll.ZZRQ,'yyyy-MM-dd') as ZZRQ," +
                            "ll.SCCJ as SCCJ," +
                            "cast(round(zx1.LJZX) as STRING)  LJZXGL," +
                            "cast(round(zx2.LJZX) as STRING)  GJXHZX," +
                            "cast(round(zx3.LJZX) as STRING)  ZJXHZX," +
                            "jxxx.data as JXXX," +
                            "pgcs.data as PGCSZB," +
                            "'' as  XHHJFB," +
                            "'' as LSGZFX," +
                            //"case when ztpg.pgjg is null then 'A' WHEN  ztpg.pgjg = 'B' THEN 'C' WHEN ztpg.pgjg = 'C' THEN 'D' else '' END as ZHPJ_ZTZL," +
                            "case when ztpg.pgjg is null then 'A' else ztpg.pgjg END as ZHPJ_ZTZL," +
                            "date_format(now(),'yyyy-MM-dd hh:mm:ss') as UPDATATIME," +
                            "gxzd.WZMC as WZ," +
                            "'牵引系统' AS SysFL ," +
                            "'012204' AS SysFLCode1," +
                            "'PJ01308' AS SysFLCode2," +
                            "date_format(jxdt_mid.JXRQ,'yyyy-MM-dd') as JXRQ," +
                            "'' as GJCSQXT," +
                            "ps.lj as LJ," +
                            "ps.LJ_ID as LJ_ID," +
                            "ps.ljjc as LJJC," +
                            "ps.psd as PSD," +
                            "ps.PSD_ID as PSD_ID," +
                            "ps.psdjc as JWDJC," +
                            " ps.ljpxm as LJPXM FROM " +
                            "(select * from ps_cx_dllx_lj_jwd where CX in ('HXD1D','HXD1C')) ps  " +     //配属相关
                            "left join (select SYCX,LBJBM,WZMC from dim_jmis_t_jcbm_jcgxzd where SYCX in ('HXD1D','HXD1C') AND LBJBM = 'PJ00003') gxzd on ps.cx = gxzd.SYCX " +                   //构型字典
                            "left join (SELECT * FROM dim_jmis_jcda_jcll_1 WHERE PJFLBM = 'PJ00003') ll on ps.s_train_id = ll.s_train_id " +                                           //机车履历
                            "left join ztpg on ps.s_train_id = ztpg.s_train_id and gxzd.WZMC = ztpg.BJWZ " +                                                                           //整体评估
                            "left join pgcs on ps.s_train_id = pgcs.s_train_id and gxzd.WZMC = pgcs.WZMC " +                                                                           //评估参数
                            "left join jxxx on ps.s_train_id = jxxx.s_train_id and gxzd.WZMC = jxxx.WZ " +                                                                             //检修信息
                            "left join zx1 on ps.s_train_id = zx1.S_TRAIN_id " +                                                                                                       //走行数据
                            "left join zx2 on ps.s_train_id = zx2.S_TRAIN_id " +                                                                                                       //高级修后走行数据
                            "left join zx3 on ps.s_train_id = zx3.S_TRAIN_id " +                                                                                                       //最近修后走行数据
                            "left join jxdt_mid on ps.s_train_id = jxdt_mid.s_train_id";                                                                                               //检修动态，中大修后日期

                    ss.sql(sqlre).write().mode("OverWrite").format("org.apache.phoenix.spark")
                            .option("dirver", "org.apache.phoenix.jdbc.PhoenixDriver")
                            .option("zkUrl", "jdbc:phoenix:JWGL-YY-T141149,JWGL-YY-T141150,JWGL-YY-T141151,JWGL-YY-T141152,JWGL-YY-V139150,JWGL-YY-V139151,JWGL-YY-V139152,JWGL-YY-V139153:2181:/hbase-unsecure")
                            .option("table", "LOCOMOTIVE_ADS.PARTS_LABEL").save();

                }

            }
        });


        jsc.start();
        try {
            jsc.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
